package com.zhxy.util.message;

import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import com.zhxy.util.async.ResultHandler;

/**
 * 维护每个管道的请求和响应管道
 * @author 刘勉
 *
 */
public class PipelineSession {
	
	private long timeout;

	/** 回复队列 */
	private Queue<ResponseHandler> responsePipeline = new LinkedBlockingQueue<ResponseHandler>();

	/** 请求队列 */
	private Queue<RequestHolder> requestPipeline = new LinkedBlockingQueue<RequestHolder>();

	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

	public Queue<ResponseHandler> getReplyQueue() {
		return responsePipeline;
	}
	
	public void addRequest(ResultHandler<Object> handler) {
		RequestHolder holder = new RequestHolder();
		holder.handler = handler;
		holder.timestamp = System.currentTimeMillis();
		requestPipeline.add(holder);
	}
	
	public ResultHandler<Object> removeRequest() {
		if (requestPipeline.peek() == null) {
			throw new IllegalStateException("请求管道已经为空,可能是出了问题");
		}
		return requestPipeline.remove().handler;
	}
	
	public ResultHandler<Object> newReponseHandler() {
		ResponseHandler handler = new ResponseHandler();
		responsePipeline.add(handler);
		return handler;
	}
	
	public void close(Throwable e) {
		for (RequestHolder holder : requestPipeline) {
			holder.handler.exception(e);
		}
		requestPipeline.clear();
		responsePipeline.clear();
	}
	
	public boolean checkRequestQueueTimeout(long now) {
		RequestHolder holder = requestPipeline.peek();
		if (holder == null)
			return false;
		else if (holder.timestamp + timeout < now) {
			throw new TimeoutException();
		} else {
			return true;
		}
	}
	
	public Object checkReponseQueue(long now) {

		Queue<ResponseHandler> queue = responsePipeline;
		ResponseHandler tail = queue.peek();
		if (tail == null) {
			return null;
		} else if (tail.done) {
			queue.remove();
			return tail.result;
		} else if (tail.timestamp + timeout < now) {
			queue.remove();
			throw new TimeoutException();
		}
		return null;
	}
	
	class RequestHolder {
		
		private volatile long timestamp = System.currentTimeMillis();
		
		private ResultHandler<Object> handler;
	}

	class ResponseHandler implements ResultHandler<Object> {

		private volatile long timestamp = System.currentTimeMillis();

		private volatile Object result;

		private volatile boolean done;

		@Override
		public void result(Object result) {
			this.result = result;
			done = true;
		}

		@Override
		public void exception(Throwable e) {
			throw new UnsupportedOperationException();
		}
	}
}
